package com.bbx.demo.controller;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;

@Slf4j
@RequestMapping("/HDFS")
@RestController
public class HDFSController {


    private String uri = "hdfs://localhost:9000";


    /**
     * 创建目录
     * @throws IOException
     * @throws URISyntaxException
     * @throws InterruptedException
     */
    @RequestMapping("/testMkdir")
    public void testMkdir() throws IOException, URISyntaxException, InterruptedException {
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        // 配置在集群上运行
        // configuration.set("fs.defaultFS", "hdfs://hadoop102:9000");
        // FileSystem fs = FileSystem.get(configuration);
        FileSystem fs = FileSystem.get(new URI("hdfs://localhost:9000"), configuration, "root");
        // 2 创建目录
        fs.mkdirs(new Path("/1108/daxian/banzhang"));
        // 3 关闭资源
        fs.close();

    }


    /**
     *文件上传
     * @param localPath
     * @param fsPath
     * @throws Exception
     */
    @RequestMapping("/uploadFile")
    public void  uploadFile(String localPath,String fsPath) throws Exception {
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI(uri), configuration, "root");
        fs.copyFromLocalFile(new Path(localPath),new Path(fsPath));
        fs.close();
    }


    /**
     * 下载文件
     * @param localPath
     * @param fsPath
     * @throws Exception
     */
    @RequestMapping("/downloadFile")
    public void downloadFile(String localPath,String fsPath) throws Exception {
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI(uri), configuration, "root");
//        fs.copyToLocalFile(new Path(fsPath),new Path(localPath));
        /**
         * 	boolean delSrc 指是否将原文件删除
         * 	Path src 指要下载的文件路径
         * 	Path dst 指将文件下载到的路径
         * 	boolean useRawLocalFileSystem 是否开启文件校验
         */
        fs.copyToLocalFile(false,new  Path(fsPath),new Path(localPath),true);
        fs.close();

    }

    /**
     *删除文件
     * @param fsPath   要删除的目录和文件
     * @throws Exception
     */
    @RequestMapping("/delFile")
    public  void  delFile(String fsPath) throws Exception {
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI(uri), configuration, "root");
        /**
         * Path path    hdfs 要删除的中目标路径
         * boolean b  是否递归删除
         */
        fs.delete(new Path(fsPath), true);
        fs.close();
    }

    /**
     *文件名更改
     * @param fsPath   要更改名称的文件
     * @param dstPath  更改后的文件
     * @throws Exception
     */
    @RequestMapping("/rename")
    public void rename(String fsPath,String dstPath) throws Exception {
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI(uri), configuration, "root");
        fs.rename(new Path(fsPath), new Path(dstPath));
        fs.close();
    }

    /**
     * HDFS 文件详情查看
     * @param fsPath
     * @throws Exception
     */
    @RequestMapping("/lsFile")
    public void  lsFile(String fsPath) throws Exception {
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI(uri), configuration, "root");
        RemoteIterator<LocatedFileStatus> remoteIterator = fs.listFiles(new Path(fsPath), true);
        while ( remoteIterator.hasNext()) {
            LocatedFileStatus fileStatus = remoteIterator.next();
            log.info("文件名：[{}]", fileStatus.getPath().getName());
            log.info("文件长度：[{}]", fileStatus.getLen());
            log.info("文件权限：[{}]", fileStatus.getPermission());
            log.info("文件组：[{}]", fileStatus.getGroup());
            log.info("文件所有者：[{}]", fileStatus.getOwner());
            log.info("文件访问时间：[{}]", fileStatus.getAccessTime());
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            for (BlockLocation blockLocation : blockLocations) {
                String[] hosts = blockLocation.getHosts();
                Arrays.stream(hosts).forEach(i->log.info("文件块所属Host:[{}]",i));
            }
            log.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
        }
    }


    /**
     * 判断HDFS 是文件还是文件夹
     * @param fsPath
     * @throws Exception
     */
    @RequestMapping("/listStatus")
    public void listStatus(String fsPath) throws  Exception{
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI(uri), configuration, "root");
        FileStatus[] fileStatuses = fs.listStatus(new Path(fsPath));
        Arrays.stream(fileStatuses).forEach(i->{
            if (i.isFile()) {
                log.info("文件名为[{}]", i.getPath().getName());
            }
            if (i.isDirectory()) {
                log.info("文件目录为[{}]", i.getPath().getName());
            }
        });

    }


}
